package com.match.match;

import java.math.BigDecimal;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.map.IMap;
import com.hazelcast.query.Predicate;
import com.hazelcast.query.Predicates;
import com.match.common.model.dto.LevelMatch;
import com.match.common.model.dto.MatchOrder;
import com.match.common.model.dto.Trade;
import com.match.constant.DealWay;
import com.match.constant.OrderState;
import com.match.plugin.hazelcast.HazelcastKit;
import com.match.plugin.kafka.Kafka;

/**
 * @ClassName: MatchDetailHandler
 * @Description: TODO(撮合细节过程，深度处理，发送数据)
 */
public class MatchDetailHandler {
	
    private static MatchDetailHandler matchDetailHandler = new MatchDetailHandler();

	public static MatchDetailHandler getMatchDetailHandler(){
        return matchDetailHandler;
    }
	
	/**
	 * 这里是修改被吃到的订单详情
	 * @Title: outMatchDepth 保证了原子操作，无需事务(异步)
	 * @Description: TODO(out订单处理)
	 * @param @param order 入单
	 * @return void 返回类型
	 * @throws
	 */
	public void outMatchDepth(MatchOrder order,HazelcastInstance hzInstance) {
		List<LevelMatch> list = order.getList();//获取之前记录的每一次撮合操作的价格和数量和是否有剩余
		try {
			if (null!=list&&list.size()>0) {
				Iterator<LevelMatch> itr = list.iterator();
				while (itr.hasNext()){//遍历每次撮合记录
					LevelMatch lm = itr.next();
					itr.remove();
					BigDecimal dealNumber = lm.getNumber();
					while (dealNumber.compareTo(BigDecimal.ZERO)>0) {
						//对手盘（获取对应交易对你（买/卖）的map）
						IMap<Long, MatchOrder> order_map = hzInstance.getMap(HazelcastKit.getOrderBookKey(order.getCoinTeam(), !order.getIsBuy()));

						Predicate<Long, MatchOrder> pricePredicate = Predicates.equal("price", lm.getPrice());
						//相同价格，时间优先，（通过流处理）
						MatchOrder queueOrder = order_map.values(pricePredicate).parallelStream().min(Comparator.comparing(MatchOrder::getCreateTime)).get();
						MatchOrder out = order_map.remove(queueOrder.getId());//先移除，返回被移除的
						if (null!=out) {
							int cpr = dealNumber.compareTo(out.getUnFinishNumber());
							if (cpr>0) {
								dealNumber=dealNumber.subtract(out.getUnFinishNumber());
								this.updateOutOder(out, OrderState.FINISH_DEAL, out.getUnFinishNumber());
							}else if (cpr==0) {
								this.updateOutOder(out, OrderState.FINISH_DEAL, dealNumber);
								dealNumber = BigDecimal.ZERO;
								break;
							}else {
								out = this.updateOutOder(out, OrderState.SOME_DEAL, dealNumber);
								order_map.put(out.getId(), out);
								dealNumber = BigDecimal.ZERO;
								break;
							}
						}
					}
				}
			}
			order.setList(null);//清空吃单记录，减小内存。
		} catch (Exception e) {
			e.printStackTrace();
			System.out.println("===出单数据处理异常,数据原型："+order.toJsonString()+"   本次异常："+e);
		}
	}
	
	
	/**
	 * 这里是发送被吃单的变化情况
	 * @Title: updateOutOder
	 * @Description: TODO(这里用一句话描述这个方法的作用)
	 * @param  out 出订单
	 * @param  dealNumber 交易数量
	 * @return void 返回类型
	 * @throws
	 */
	private MatchOrder updateOutOder(MatchOrder out,OrderState orderState ,BigDecimal dealNumber) {
		out.setState(orderState.value);
		out.setFinishNumber(out.getFinishNumber().add(dealNumber));
		out.setUnFinishNumber(out.getUnFinishNumber().subtract(dealNumber));
		if (out.getIsBuy()) {
			out.setSurplusFrozen(out.getSurplusFrozen().subtract(dealNumber.multiply(out.getPrice())));
		}else {
			out.setSurplusFrozen(out.getSurplusFrozen().subtract(dealNumber));
		}
		Kafka.send(Kafka.orderChange, Kafka.orderChange, out.getId(), out);
		Kafka.send(Kafka.matchTrade, Kafka.matchTrade, out.getId(), new Trade()
				.setUid(out.getUid())
				.setOid(out.getId())
				.setIsBuy(out.getIsBuy())
				.setNumber(dealNumber)
				.setPrice(out.getPrice())
				.setTotalPrice(dealNumber.multiply(out.getPrice()))
				.setCoinTeam(out.getCoinTeam())
				.setDealWay(DealWay.MAKER.value));
//		sendOrderChange(out);
//		ThreadPoolUtil.getExec().submit(new Runnable() {
//			public void run() {
//				sendTradeRecord(out, out.getPrice(), dealNumber, DealWay.MAKER);
//			}
//		});
		return out;
	}
}
